package ods

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.util.Properties

//todo 脱敏

object TransODS {
  val spark: SparkSession = SparkSession.builder()
    .master("spark://hadoop01:7077")
    .enableHiveSupport()
    .config("spark.sql.parquet.writeLegacyFormat","true")
    .getOrCreate()
  spark.catalog.setCurrentDatabase("mall")

  def changeFieldIf(field: String, target: String, value: String) = if (field.matches(s".*$target")) value else field

  def trans(from: String, to: String)(sql: DataFrame => DataFrame) = {
    val url = "jdbc:mysql://hadoop01:3306/mall?useUnicode=true&characterEncoding=utf-8&useSSL=false"
    val prop = new Properties()
    prop.put("driver","com.mysql.jdbc.Driver")
    prop.put("user","hive")
    prop.put("password","123456")
    val df = spark.read.jdbc(url, from, prop)
    sql(df).write.mode(SaveMode.Overwrite).insertInto(to)
  }

  def main(args: Array[String]): Unit = {

    val incDate = "2024-05-01"

    trans("activity_info", "ods_activity_info_full") {
      _.selectExpr("id", "date_format(create_time,'yyyy-MM-dd') as k1",  "activity_name", "activity_type", "activity_desc", "start_time", "end_time", "create_time")
    }

    trans("activity_rule", "ods_activity_rule_full") {identity}

    trans("base_category1", "ods_base_category1_full")  { identity }
    trans("base_category2", "ods_base_category2_full")  { identity }
    trans("base_category3", "ods_base_category3_full")  { identity }

    trans("base_dic","ods_base_dic_full") {
      _.selectExpr("dic_code",
        "dic_name",
        "date_format(create_time,'yyyy-MM-dd') as k1",
        "parent_code",
        "create_time",
        "operate_time"
      )
    }

    trans("base_province","ods_base_province_full") {identity}

    trans("base_region", "ods_base_region_full") {identity}

    trans("base_trademark","ods_base_trademark_full") {identity}

    trans("cart_info", "ods_cart_info_full") {
      _.selectExpr(
        "id",
        "date_format(create_time,'yyyy-MM-dd') as k1",
        "user_id", "sku_id", "cart_price", "sku_num", "img_url", "sku_name", "is_checked", "create_time", "operate_time", "is_ordered", "order_time", "source_type", "source_id"
      )
    }

    trans("cart_info", "ods_cart_info_inc") {
      _.selectExpr(
        "id",
        "date_format(create_time,'yyyy-MM-dd') as k1",
        "user_id", "sku_id", "cart_price", "sku_num", "img_url", "sku_name", "is_checked", "create_time", "operate_time", "is_ordered", "order_time", "source_type", "source_id"
      ).where(s"create_time >= date('$incDate')")
    }

    trans("comment_info", "ods_comment_info_inc") {
      _.selectExpr(
        "id", "date_format(create_time,'yyyy-MM-dd') as k1","user_id", "nick_name", "head_img", "sku_id", "spu_id", "order_id", "substr(appraise, -1, 1) as appraise", "comment_txt", "create_time", "operate_time"
      ).where(s"create_time >= date('$incDate') and appraise between 1 and 3 ")
    }


    trans("coupon_info", "ods_coupon_info_full") {
      _.selectExpr("id, k1, coupon_name, coupon_type, condition_amount, condition_num, activity_id, benefit_amount, benefit_discount, create_time, range_type, limit_num, taken_count, start_time, end_time, operate_time, expire_time"
        .split(",").map{changeFieldIf(_, "range_type", "substr(range_type, -1, 1)  as range_type")}
        .map{changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
        : _*
      ).where("range_type between 1 and 3")
    }

    trans("coupon_use","ods_coupon_use_inc") {
      _.selectExpr(
        "id,k1, coupon_id, user_id, order_id, coupon_status, create_time, get_time, using_time, used_time, expire_time"
          .split(",").map {changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      ).where(s"create_time >= date('$incDate')").drop("create_time")
    }

    trans("favor_info", "ods_favor_info_inc") {
      _.selectExpr(
        "id, k1,  user_id, sku_id, spu_id, is_cancel, create_time, cancel_time".split(",")
          .map {changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
        :_*
      )
    }

    trans("order_detail_activity", "ods_order_detail_activity_inc") {
      _.selectExpr(
        "id, k1, order_id, order_detail_id, activity_id, activity_rule_id, sku_id, create_time".split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
        :_*
      ).where(s"create_time >= date('$incDate')")
    }

    trans("order_detail_coupon", "ods_order_detail_coupon_inc") {
      _.selectExpr(
        "id, k1,  order_id, order_detail_id, coupon_id, coupon_use_id, sku_id, create_time".split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
        :_*
      ).where(s"create_time >= date('$incDate')")
    }


    trans("order_detail", "ods_order_detail_inc") {
      _.selectExpr(
        "id, k1, order_id, sku_id, sku_name, img_url, order_price, sku_num, create_time, source_type, source_id, split_total_amount, split_activity_amount, split_coupon_amount".split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      ).where(s"create_time >= date('$incDate')")
    }

    trans("order_info", "ods_order_info_inc") {
      _.selectExpr(
          "id, k1, consignee, consignee_tel, total_amount, order_status, user_id, payment_way, delivery_address, order_comment, out_trade_no, trade_body, create_time, operate_time, expire_time, process_status, tracking_no, parent_order_id, img_url, province_id, activity_reduce_amount, coupon_reduce_amount, original_total_amount, feight_fee, feight_fee_reduce, refundable_time".split(",")
            .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
            :_*
      ).where(s"create_time >= date('$incDate')")
    }

    trans("order_refund_info", "ods_order_refund_info_inc") {
      _.selectExpr(
        "id, k1, user_id, order_id, sku_id, refund_type, refund_num, refund_amount, refund_reason_type, refund_reason_txt, refund_status, create_time"
          .split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      ).where(s"create_time >= date('$incDate')")
    }


    trans("order_status_log","ods_order_status_log_inc") {
      _.selectExpr(
        "id,k1, order_id, order_status, operate_time"
          .split(",")
          .map{ changeFieldIf(_, "k1", "date_format(operate_time,'yyyy-MM-dd') as k1")}
          :_*
      ).where(s"operate_time >= date('$incDate')")
    }

    trans("payment_info","ods_payment_info_inc") {
      _.selectExpr(
        "id,k1, out_trade_no, order_id, user_id, payment_type, trade_no, total_amount, subject, payment_status, create_time, callback_time, callback_content"
          .split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      ).where(s"create_time >= date('$incDate')")
    }

    trans("refund_payment","ods_refund_payment_inc") {
      _.selectExpr(
        "id, k1, out_trade_no, order_id, sku_id, payment_type, trade_no, total_amount, subject, refund_status, create_time, callback_time, callback_content "
          .split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      )
    }


    trans("sku_attr_value", "ods_sku_attr_value_full") {identity}

    trans("sku_info","ods_sku_info_full") {
      _.selectExpr(
        "id,k1,spu_id, price, sku_name, sku_desc, weight, tm_id, category3_id, sku_default_img, is_sale, create_time"
          .split(",")
          .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
          :_*
      )
    }

    trans("sku_sale_attr_value", "ods_sku_sale_attr_value_full") {identity}

    trans("spu_info", "ods_spu_info_full") {identity}

    trans("user_info", "ods_user_info_inc") {
      _.selectExpr(
        "id,k1, login_name, nick_name, passwd, name, phone_num, email, head_img, user_level, birthday, gender, create_time, operate_time, status "
       .split(",")
        .map{ changeFieldIf(_, "k1", "date_format(create_time,'yyyy-MM-dd') as k1")}
        :_*
      )
    }


  }
}
